package com.atguigu.gmall.realtime.app.dwd;

import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import com.atguigu.gmall.realtime.util.MySqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DwdTradeOrderCancelDetail {



        // 0 环境
        //  1  获得topic_db
        //  2  分别创建四个动态表
        //2.1   订单主表      order_info
        //2.2   订单明细表  order_detail
        //2.3   购物券  order_detail_coupon
        //2.4   活动    order_detail_activity
        // 3   字典维度表
        //4   把五张表 进行join操作
        // 5  目标表的定义
        // 6   写入目标表
        public static void main(String[] args) {
            // 0 环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

            StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

            tableEnv.getConfig().setIdleStateRetention(Duration.ofMillis(20*60000));

            //  1  获得topic_db
            String createTopicDbSQL="     CREATE TABLE topic_db (\n" +
                    "        `database` STRING,\n" +
                    "       `table` STRING,\n" +
                    "        `type` STRING,\n" +
                    "         `data` MAP<STRING,STRING>,\n" +
                    "         `old` MAP<STRING,STRING> ,\n" +
                    "          `proc_time`  as proctime(), \n"+
                    "         ts   STRING \n"+
                    "          ) WITH (\n" +
                    "          'connector' = 'kafka',\n" +
                    "          'topic' = 'topic_db',\n" +
                    "          'properties.bootstrap.servers' = 'hadoop102:9092',\n" +
                    "          'properties.group.id' = 'dwd_trade_order_detail',\n" +
                    "          'scan.startup.mode' = 'group-offsets',\n" +
                    "           'format' = 'json'\n" +
                    "         )";
            tableEnv.executeSql(createTopicDbSQL);

            //  2  分别创建四个动态表
            //2.1   订单主表      order_info
            Table orderInfo = tableEnv.sqlQuery("select \n" +
                    "data['id'] id,\n" +
                    "data['user_id'] user_id,\n" +
                    "data['province_id'] province_id\n" +
                    "from `topic_db`\n" +
                    "where `table` = 'order_info'\n" +
                    "and `type` = 'update' and data['order_status']='1003'");
            tableEnv.createTemporaryView("order_info", orderInfo);



            //2.2   订单明细表  order_detail
            Table orderDetail = tableEnv.sqlQuery("select \n" +
                    "data['id'] id,\n" +
                    "data['order_id'] order_id,\n" +
                    "data['sku_id'] sku_id,\n" +
                    "data['sku_name'] sku_name,\n" +
                    "data['create_time'] create_time,\n" +
                    "data['source_id'] source_id,\n" +
                    "data['source_type'] source_type,\n" +
                    "data['sku_num'] sku_num,\n" +
                    "cast(cast(data['sku_num'] as decimal(16,2)) * " +
                    "cast(data['order_price'] as decimal(16,2)) as String) split_original_amount,\n" +
                    "data['split_total_amount'] split_total_amount,\n" +
                    "data['split_activity_amount'] split_activity_amount,\n" +
                    "data['split_coupon_amount'] split_coupon_amount,\n" +
                    "ts,\n" +
                    "proc_time\n" +
                    "from `topic_db` where `table` = 'order_detail' " +
                    "and `type` = 'insert'\n");
            tableEnv.createTemporaryView("order_detail", orderDetail);


            //2.3   购物券  order_detail_coupon
            Table orderDetailCoupon = tableEnv.sqlQuery("select\n" +
                    "data['order_detail_id'] order_detail_id,\n" +
                    "data['coupon_id'] coupon_id\n" +
                    "from `topic_db`\n" +
                    "where `table` = 'order_detail_coupon'\n" +
                    "and `type` = 'insert'\n");
            tableEnv.createTemporaryView("order_detail_coupon", orderDetailCoupon);

            //2.4   活动    order_detail_activity
            Table orderDetailActivity = tableEnv.sqlQuery("select \n" +
                    "data['order_detail_id'] order_detail_id,\n" +
                    "data['activity_id'] activity_id,\n" +
                    "data['activity_rule_id'] activity_rule_id\n" +
                    "from `topic_db`\n" +
                    "where `table` = 'order_detail_activity'\n" +
                    "and `type` = 'insert'\n");
            tableEnv.createTemporaryView("order_detail_activity", orderDetailActivity);

            // 3   字典维度表
            String baseDicDDL = MySqlUtil.getBaseDicDDL();
            tableEnv.executeSql(baseDicDDL);

            //4   把五张表 进行join操作
            Table resultTable = tableEnv.sqlQuery("select \n" +
                    "od.id,\n" +
                    "od.order_id,\n" +
                    "oi.user_id,\n" +
                    "od.sku_id,\n" +
                    "od.sku_name,\n" +
                    "oi.province_id,\n" +
                    "act.activity_id,\n" +
                    "act.activity_rule_id,\n" +
                    "cou.coupon_id,\n" +
                    "date_format(od.create_time, 'yyyy-MM-dd') date_id,\n" +
                    "od.create_time,\n" +
                    "od.source_id,\n" +
                    "od.source_type,\n" +
                    "dic.dic_name source_type_name,\n" +
                    "od.sku_num,\n" +
                    "od.split_original_amount,\n" +
                    "od.split_activity_amount,\n" +
                    "od.split_coupon_amount,\n" +
                    "od.split_total_amount,\n" +
                    "od.ts \n" +
                    "from order_detail od \n" +
                    "join order_info oi\n" +
                    "on od.order_id = oi.id\n" +
                    "left join order_detail_activity act\n" +
                    "on od.id = act.order_detail_id\n" +
                    "left join order_detail_coupon cou\n" +
                    "on od.id = cou.order_detail_id\n" +
                    "join `base_dic` for system_time as of od.proc_time as dic\n" +
                    "on od.source_type = dic.dic_code");
            tableEnv.createTemporaryView("result_table", resultTable);

            //   tableEnv.executeSql("select * from "+resultTable).print();

            // 5  目标表的定义
            tableEnv.executeSql("" +
                    "create table dwd_trade_order_cancel_detail(\n" +
                    "id string,\n" +
                    "order_id string,\n" +
                    "user_id string,\n" +
                    "sku_id string,\n" +
                    "sku_name string,\n" +
                    "province_id string,\n" +
                    "activity_id string,\n" +
                    "activity_rule_id string,\n" +
                    "coupon_id string,\n" +
                    "date_id string,\n" +
                    "create_time string,\n" +
                    "source_id string,\n" +
                    "source_type string,\n" +
                    "source_type_name string,\n" +
                    "sku_num string,\n" +
                    "split_original_amount string,\n" +
                    "split_activity_amount string,\n" +
                    "split_coupon_amount string,\n" +
                    "split_total_amount string,\n" +
                    "ts string,\n" +
                    "primary key(id) not enforced\n" +
                    ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_order_cancel_detail"));


            // 6   写入目标表

            tableEnv.executeSql("" +
                    "insert into dwd_trade_order_cancel_detail \n" +
                    "select * from result_table");


    }
}
